package com.tengju.user.infrastructure.persistence.es;

import com.google.common.collect.Lists;
import com.tengju.support.infrastructure.aop.MonitorTime;
import com.tengju.user.application.user.UserDocumentService;
import com.tengju.user.application.user.dto.UserSearchForm;
import com.tengju.user.domain.annotation.ElasticDecrypt;
import com.tengju.user.domain.annotation.ElasticEncrypt;
import com.tengju.user.domain.model.doc.UserDocument;
import com.tengju.user.domain.model.user.UserIdCode;
import com.tengju.user.domain.service.UserDocumentRepository;
import com.tengju.user.infrastructure.shared.InfraException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.index.query.QueryBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchScrollHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Repository;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Repository
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class UserDocumentServiceImpl implements UserDocumentService {

    private final UserDocumentRepository userDocumentRepository;
    private final ElasticsearchRestTemplate elasticsearchTemplate;

    @Override
    @ElasticDecrypt
    @ElasticEncrypt
    public Page<UserDocument> search(UserSearchForm form) {

        QueryBuilder queryBuilder = UserSearchDocConvert.formToQuery(form);
        Pageable pageable = PageRequest.of(form.getPageIndex() <= 0 ? 0 :
                form.getPageIndex() - 1, form.getPageSize());

        NativeSearchQuery searchQueryBuilder = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder).withPageable(pageable).build();

        return userDocumentRepository.search(searchQueryBuilder);
    }

    @Override
    @ElasticDecrypt
    public List<UserDocument> findByIds(List<UserIdCode> userIdCodes) {
        List<UserDocument> result = Lists.newArrayList();
        if (CollectionUtils.isEmpty(userIdCodes)) {
            return result;
        }
        List<Long> idList = userIdCodes.stream().map(UserIdCode::getValue).collect(Collectors.toList());
        Iterator<UserDocument> iterator = userDocumentRepository.findAllById(idList).iterator();
        while (iterator.hasNext()) {
            result.add(iterator.next());
        }
        return result;
    }

    @Override
    @ElasticDecrypt
    public UserDocument findById(UserIdCode userIdCode) {
        if (userIdCode == null || userIdCode.getValue() <= 0) {
            return null;
        }
        return userDocumentRepository.findById(userIdCode.getValue()).orElse(null);
    }

    @Override
    @ElasticEncrypt
    public void saveAll(Iterable<UserDocument> entities) {
        userDocumentRepository.saveAll(entities);
    }

    @Override
    public void deleteByIdCode(UserIdCode idCode) {
        userDocumentRepository.deleteById(idCode.getValue());
    }

    @Override
    @MonitorTime
    @ElasticDecrypt
    @ElasticEncrypt
    public Map<String/* scrollId */, List<UserDocument>> searchScroll(UserSearchForm form) {
        List<UserDocument> result = Lists.newArrayList();
        SearchScrollHits<UserDocument> searchResponse = null;
        String nextScrollId = "";
        Integer size = form.getPageSize();
        if (size > 1000) {
            throw new InfraException("单页查询数据量太大");
        }

        try {
            if (StringUtils.isNotBlank(form.getScrollId())) {
                // 初始化scroll查询
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(form.getScrollId());
                searchScrollRequest.scroll("5m");
                // 发起请求并接收响应
                searchResponse = elasticsearchTemplate.searchScrollContinue(form.getScrollId(),
                        10000,
                        UserDocument.class,
                        IndexCoordinates.of("user_doc"));
            } else {
                QueryBuilder queryBuilder = UserSearchDocConvert.formToQuery(form);
                NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
                        .withQuery(queryBuilder)
                        .withPageable(PageRequest.of(0, size))
                        .build();
                // 发起请求并接收响应
                searchResponse = elasticsearchTemplate.searchScrollStart(10000,
                        searchQuery, UserDocument.class,
                        IndexCoordinates.of("user_doc"));
            }
            // 获取ScrollId
            nextScrollId = searchResponse.getScrollId();
            if (searchResponse.hasSearchHits()) {
                searchResponse.getSearchHits().forEach(s -> {
                    result.add(s.getContent());
                });
            }
        } catch (Exception e) {
            log.error("search scroll error", e);
            throw new InfraException("search scroll error" , e);
        }
//        finally {
//            if (searchResponse != null && searchResponse.hasSearchHits()) {
//                elasticsearchTemplate.searchScrollClear(Lists.newArrayList(nextScrollId));
//            }
//        }
        if (CollectionUtils.isEmpty(result)) {
            return Collections.emptyMap();
        }
        return Collections.singletonMap(nextScrollId, result);
    }

    /**
     * 查询全量数据，谨慎使用，一般用于数据导出
     * 不超过5w条
     * @param form
     * @return
     */
    @Override
    @MonitorTime
    @ElasticDecrypt
    @ElasticEncrypt
    public List<UserDocument> batchSearchScroll(UserSearchForm form) {
        List<UserDocument> result = new ArrayList<>();
        int loop = 1;
        form.setPageSize(1000);
        while (loop <= 50) {
            Map<String, List<UserDocument>> resultMap = searchScroll(form);
            if (CollectionUtils.isEmpty(resultMap)) {
                break;
            }
            List<UserDocument> scrollList = resultMap.values().stream()
                    .flatMap(Collection::stream)
                    .collect(Collectors.toList());
            result.addAll(scrollList);
            String nextScrollId = resultMap.keySet().stream()
                    .findFirst()
                    .orElseGet(() -> {
                        throw new InfraException("es scroll查询异常，没有正常返回scrollId");
                    });
            if (StringUtils.isBlank(nextScrollId)) {
                break;
            }
            form.setScrollId(nextScrollId);
            loop++;
        }
        return result;
    }
}
